package com.ydl.learning.flink.demo.pipeline

import com.ydl.learning.flink.demo.pipeline.java.entity.JobConfig
import org.apache.flink.streaming.api.scala.DataStream


/**
 *
 *
 * @author ydl
 * @since 2022/1/7
 */
class Pipeline2 {
  private var jobConfig: JobConfig = _
  private var source: Source[PipelineData] = _
  private var processes: List[Process[PipelineData]] = _
  private var sink: Sink[PipelineData] = _

  def this(builder: Builder2) = {
    this
    this.jobConfig = builder.jobConfig
    this.source = builder.source
    this.processes = builder.processes
    this.sink = builder.sink

  }

  def execute(): Unit = {
    source.setConfig(this.jobConfig)
    source.checkConfig()
    var result: DataStream[PipelineData] = source.init()
    for (process <- processes) {
      process.setData(result)
      result = process.run
    }
    sink.setSource(result)
    sink.sink()
    source.env().execute(jobConfig.getJobName)
  }


}